package com.mjf.hotitems_analysis


import java.sql.Timestamp

import com.mjf.dim.UserBehavior
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, Slide, Table}
import org.apache.flink.table.api.scala._

object HotItemsWithTable {
  def main(args: Array[String]): Unit = {

    // 创建流处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 定义时间语义
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 从文件读取数据
    val inputStream: DataStream[String] = env.readTextFile("D:\\coding\\idea\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv")

    // 将数据转换为样例类，并且提取timestamp定义watermark
    val dataStream: DataStream[UserBehavior] = inputStream.map {
      line =>
        val dataArray: Array[String] = line.split(",")
        UserBehavior(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt, dataArray(3), dataArray(4).toLong)
    }.assignAscendingTimestamps(_.timestamp * 1000L)  // 数据是有序的，不需要定义延迟处理乱序数据

    // 要调用TableApi,先创建表执行环境
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)

    // 将DataStream转换成表，提取需要的字段进行处理
    val dataTable: Table = tableEnv.fromDataStream(dataStream, 'itemId, 'behavior, 'timestamp.rowtime as 'ts)

    // 分组开窗增量聚合
    val aggTable: Table = dataTable
      .filter('behavior === "pv")
      .window(Slide over 1.hours every 5.minutes on 'ts as 'sw)
      .groupBy('itemId, 'sw)
      .select('itemId, 'itemId.count as 'cnt, 'sw.end as 'windowEnd)

    // 用SQL实现分组选取TopN的功能
    tableEnv.createTemporaryView("agg", aggTable, 'itemId, 'cnt, 'windowEnd)

    val resultTable: Table = tableEnv.sqlQuery(
      """
        |select
        |   *
        |from
        |   (
        |   select
        |       *,
        |       row_number() over(partition by windowEnd order by cnt desc) as rn
        |   from
        |       agg
        |   ) as a
        |where
        |   rn <= 5
        |""".stripMargin)

    resultTable.toRetractStream[(Long, Long, Timestamp, Long)].print("result")

    env.execute("HotItemsWithTable")

  }
}
